package com.ylkj;


import com.hwl.datahub.DataHubReaderConfig;

import com.hwl.datahub.SchemaVo;
import com.hwl.datahub.in.ProxyDatahubSourceFunction;


import com.hwl.redis.RedisConfig;
import com.hwl.utils.ConfigUtils;
import com.ylkj.bus.MessageContent;
import com.ylkj.bus.dispatcher.DispatcherHandler;

import com.ylkj.dao.RdsConfig;
import com.ylkj.dao.RedisConfigInfo;
import com.ylkj.process.UnpackConvertProcess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.IOException;
import java.io.InputStream;
import java.io.StringReader;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * @Author: wengzx
 * @Date: 2025/2/15
 * @Desc:
 */
@Slf4j
public class Application {
    public static void main(String[] args) throws Exception {
        //获取Blink配置
        Properties properties = getProperties(args);
        //获取datahub
        DataHubReaderConfig dataHubReaderConfig = ConfigUtils.getDataHubReaderConfig("source", properties);
        //执行环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        List<String> handlers = getHandlers(properties);

        //获取source
        String topicName = dataHubReaderConfig.getTopicName();
        DataHubReaderConfig clone = dataHubReaderConfig.clone();
        clone.setInnerComment(String.join("_", dataHubReaderConfig.getInnerComment(), StringUtils.join(handlers, "_")));
        clone.setTopicName(topicName);
        ProxyDatahubSourceFunction datahubSourceFunction = new ProxyDatahubSourceFunction(
                clone,
                clone.getTimestamp(),
                12L,
                1000L,
                8192
        );
        DataStream<List<SchemaVo>> sourceStream = environment.addSource(datahubSourceFunction).name(topicName + "_source");
        //转换流
        DataStream<MessageContent> unpackStream = sourceStream.process(new UnpackConvertProcess()).name("unpack");

        for (String handler : handlers) {
            DispatcherHandler dispatcherHandler = DispatcherHandler.valueOf(handler);
            dispatcherHandler.dispatch(unpackStream, properties);
        }
        environment.execute("t_iot_device_online_status_report_data");
    }

    private static Properties getProperties(String[] args) throws IOException {
        Properties flinkProperties = new Properties();
        try {
            final ParameterTool params = ParameterTool.fromArgs(args);/*此处必须写configFile，读取configFile中的参数。*/
            String configFilePath = params.get("configFile");
            /*创建一个Properties对象，用于保存在平台中设置的相关参数值。*/
            /*将平台页面中设置的参数值加载到Properties对象中。*/
            if (configFilePath != null) {
                try (StringReader reader = new StringReader(new String(Files.readAllBytes(Paths.get(configFilePath)), StandardCharsets.UTF_8));) {
                    flinkProperties.load(reader);
                }
            }
            log.info("flink配置:{}", flinkProperties);
        } catch (Exception e) {
            log.error("读取flink配置失败。");
            log.error(e.getMessage(), e);
        }
        /*获取参数。*/
        String targetConfig = (String) flinkProperties.get("properties.path");
        if (StringUtils.isBlank(targetConfig)) {
            targetConfig = "config-dev.properties";
        }
        Properties properties = new Properties();
        try (InputStream resourceAsStream = Application.class.getClassLoader().getResourceAsStream(targetConfig)) {
            properties.load(resourceAsStream);
        }
        for (String key : flinkProperties.stringPropertyNames()) {
            properties.setProperty(key, flinkProperties.getProperty(key));
        }
        return properties;
    }

    private static List<String> getHandlers(Properties properties) {
        return Arrays.asList(properties.getProperty("source.handler.names").split(","));
    }


    public static RdsConfig getRdsConfig(Properties properties) {
        return new RdsConfig(
                properties.getProperty("sink.rds.driver"),
                properties.getProperty("sink.rds.url"),
                properties.getProperty("sink.rds.username"),
                properties.getProperty("sink.rds.password")
        );
    }

    public static RedisConfig getRedisConfig(Properties properties) {
        return new RedisConfig(
                properties.getProperty("source.redis.host"),
                Integer.parseInt(properties.getProperty("source.redis.port", "6379")),
                properties.getProperty("source.redis.password"),
                Integer.parseInt(properties.getProperty("source.redis.database", "0")),
                properties.getProperty("source.redis.targetKey"),
                Integer.parseInt(properties.getProperty("source.redis.batchSize", "8192"))
        );
    }

    public static RedisConfigInfo getRedisConfig1(Properties properties) {
        return new RedisConfigInfo(
                properties.getProperty("sink.redis.host"),
                Integer.parseInt(properties.getProperty("sink.redis.port", "6379")),
                properties.getProperty("sink.redis.password"),
                Integer.parseInt(properties.getProperty("sink.redis.database", "0"))
        );
    }
}